package ch5;

/**
 * 使用bool变量isUse作为线程停止标记，并结合原子变量表示目标线程待处理任务数量
 * 在每次从通道中取出任务执行之前，目标线程都会先判断是否客户端发出线程停止信息
 * 并且没有剩下地待处理任务，若满足条件，自动结束本目标线程地执行
 * 这样能够防止线程中断标记被其他方法所清除导致目标线程无法停止的问题
 */

import util.Debug;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class TerminatableTaskRunner implements TaskRunnerSpec{
    protected final BlockingQueue<Runnable> channel;
    // 线程停止标记
    protected volatile boolean isUse = true;
    // 待处理任务计数器
    public final AtomicInteger reservations = new AtomicInteger(0);

    protected volatile Thread workerThread;

    public TerminatableTaskRunner(BlockingQueue<Runnable> channel) {
        this.channel = channel;
        this.workerThread = new WorkerThread();
    }


    public TerminatableTaskRunner() {
        this (new LinkedBlockingQueue<Runnable>());
    }

    @Override
    public void init() {
        final Thread t = workerThread;
        if (null != t) {
            t.start();
        }
    }

    @Override
    public void submit(Runnable task) throws InterruptedException {
        channel.put(task);
        // 生产者提交一个任务，并把待处理任务计数器加1
        reservations.incrementAndGet();
    }

    public void shutdown() {
        Debug.info("Shutting down service...");
        isUse = false;
        final Thread t = workerThread;
        if (null != t) {
            t.interrupt();
        }
    }

    class WorkerThread extends Thread {
       @Override
       public void run() {
          Runnable task = null;
          try {
              for (;;) {
                  // 线程不再被需要，且无待处理任务
                  if (!isUse && reservations.get() <= 0) {
                      break;
                  }
                  task = channel.take();
                  try {
                      task.run();
                  } catch (Throwable e) {
                      e.printStackTrace();
                  }
                  // 使待处理任务数减少1
                  reservations.decrementAndGet();
              }
          } catch (InterruptedException e) {
              workerThread = null;
          }
          Debug.info("worker thread terminated.");
       }
   }
}
